<?php
error_reporting ( 2047 );
class epoll {
    private static $socket;
    private static $connections = array ();
    private static $buffers = array ();
    private static $id = 0;
    private static $ct_last;
    private static $pids = array ();
    /** 
     * 实例化
     */
    public function __construct($port, $num = 1, $timer = array()) {
        if ($port < 1024)
            die ( "Port must be a number which bigger than 1024\n" );
        self::$socket = stream_socket_server ( 'tcp://0.0.0.0:' . $port, $errno, $errstr );
        stream_set_blocking ( self::$socket, 0 );
        $base = event_base_new ();
        $event = event_new ();
        event_set ( $event, self::$socket, EV_READ | EV_PERSIST, 'epoll::ev_accept', $base );
        event_base_set ( $event, $base );
        event_add ( $event );
        if (($num - 1) > 0)
            self::sw_spawn ( $num - 1 );
        $countTimers = count ( $timer );
        self::$pids [] = posix_getpid ();       
        while ( true ) {
            event_base_loop ( $base, EVLOOP_NONBLOCK );
            if ($timer) {
                $pid = posix_getpid ();
                $i = 0;
                foreach ( $timer as $k => $v ) {
                    $key = $i;
                    if ($i > $num)
                        $key = intval ( $i - $num * floor ( $i / $num ) );
                    if (self::$pids && key_exists ( $key, self::$pids )) {
                        if ($pid == self::$pids [$key]) {
                            eval ( $k . ';' );
                            usleep ( $v );
                            self::ev_sendall('hi body');
                        }
                    }
                    $i ++;
                }
            }
            //print_r(self::$connections );
            usleep ( 10 );
        }
    }
    public static function timer() {
    
    }
    /**
     * 接收请求
     */
    public static function ev_accept($socket, $flag, $base) {
        $connection = stream_socket_accept ( $socket );
        if (! $connection)
            return;
        stream_set_blocking ( $connection, 0 );        
        self::$id += 1;        
        $buffer = event_buffer_new ( $connection, 'epoll::ev_read', NULL, 'epoll::ev_error', self::$id );        
        event_buffer_base_set ( $buffer, $base );
        event_buffer_timeout_set ( $buffer, 30, 30 );
        event_buffer_watermark_set ( $buffer, EV_READ, 0, 0xffffff );
        event_buffer_priority_set ( $buffer, 10 );
        event_buffer_enable ( $buffer, EV_READ | EV_PERSIST );        
        //保存请求 
        self::$connections [self::$id] = $connection;        
        self::$buffers [self::$id] = $buffer;
    }
    /**
     * 客户端连接出错或退出
     */
    public static function ev_error($buffer, $error, $id) {
        event_buffer_disable ( self::$buffers [$id], EV_READ | EV_WRITE );
        event_buffer_free ( self::$buffers [$id] );
        fclose ( self::$connections [$id] );
        unset ( self::$buffers [$id], self::$connections [$id] );
        echo date ( "Y-m-d H:i:s" ) . ": client $id close total clients " . count ( self::$buffers ) . " \r\n";
    }
    /**
     * 读请求内容
     */
    public static function ev_read($buffer, $id) {
        static $ct = 0;
        static $ct1 = 0;
        
        while ( @$read = event_buffer_read ( $buffer, 256 ) ) {
            $ct += strlen ( $read );
            if ($ct - 10240 > @$ct_last) {
                $ct_last = $ct;
                //echo (ceil($ct/1024)*10)." K\n";  
                self::ev_send ( $id, posix_getpid () . 'welcome' . "\r\n" );
                //self::close ( $id );
            }
        
        }
    }
    /**
     * 发送消息到客户端
     */
    public static function ev_send($socket_id, $string) {
        $length = strlen ( $string );
        for($written = 0; $written < $length; $written += @$fwrite) {
            $fwrite = fwrite ( self::$connections [$socket_id], substr ( $string, $written ) );
            if ($fwrite === false)
                return $written;
        }
        return $written;
    }
    public static function ev_sendall($string)
    {
        print_r(self::$connections );
        foreach (self::$connections as $k=>$v){
            echo 'ffff';
            var_dump(self::ev_send($v, $string));
        }
    }
    /**
     * 关闭客户端
     */
    public static function close($socket_id, $event = null) {
        if ($event) {
            event_del ( $event );
            event_free ( $event );
        }
        stream_socket_shutdown ( self::$connections [$socket_id], STREAM_SHUT_RDWR );
        fclose ( self::$connections [$socket_id] );
        unset ( self::$buffers [$socket_id], self::$connections [$socket_id] );
    }
    /**
     * 开启进程数
     */
    public static function sw_spawn($num) {
        for($i = 0; $i < $num; $i ++) {
            $pid = pcntl_fork ();
            if (! $pid) {
                break;
            } else {
                self::$pids [] = $pid;
            }
        }
    }
}
function handler_timer($e) {
    
}
new epoll ( 8080, 4, array ('handler_timer("bb")' => '1000000' ) ); 